package com.chenjl.transaction.service.impl;

import java.util.Date;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import javax.annotation.Resource;

import org.apache.commons.lang3.time.DateUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.alibaba.fastjson.JSONObject;
import com.chenjl.transaction.dao.ExtTbRabbitmqSendLogMapper;
import com.chenjl.transaction.dao.TbRabbitmqConsumeLogMapper;
import com.chenjl.transaction.dao.TbRabbitmqSendLogMapper;
import com.chenjl.transaction.dto.MessageQueueBody;
import com.chenjl.transaction.model.TbRabbitmqConsumeLog;
import com.chenjl.transaction.model.TbRabbitmqSendLog;
import com.chenjl.transaction.service.MessageQueueService;

import lombok.extern.slf4j.Slf4j;
/**
 * 抽象MQ发送的服务，对外屏蔽MQ的特性
 * 2019-9-9 14:10:45
 * @author chenjinlong
 */
@Slf4j
@Service
public class MessageQueueServiceImpl implements MessageQueueService, InitializingBean {
	
	private BlockingQueue<Long> blockingQueue = new LinkedBlockingQueue<Long>();
	
	@Resource(name="rabbitmqAmqpTemplate")
	private RabbitTemplate rabbitTemplate;
	@Autowired
	private TbRabbitmqSendLogMapper tbRabbitmqSendLogMapper;
	@Autowired
	private ExtTbRabbitmqSendLogMapper extTbRabbitmqSendLogMapper;
	@Autowired
	private TbRabbitmqConsumeLogMapper tbRabbitmqConsumeLogMapper;
	
	
	
	@Override
	public void afterPropertiesSet() throws Exception {
		Thread thread = new Thread(new Runnable() {
			@Override
			public void run() {
				for(;;) {
					pollMessage();
				}
			}
		},"rabbitmq-poll-thread");
		
		thread.setDaemon(true);
		thread.start();
	}
	
	/**
	 * 守护线程定时从内存队列拉取数据
	 */
	private void pollMessage() {
		Long pkId = null;
		try {
			pkId = blockingQueue.poll(20,TimeUnit.SECONDS);
			
			if(pkId != null) {
				log.info("从blockingQueue获取要发送的消息....pkId:{}",pkId);
				
				TbRabbitmqSendLog tbRabbitmqSendLog = tbRabbitmqSendLogMapper.selectByPrimaryKey(pkId);
				if(tbRabbitmqSendLog == null) {
					blockingQueue.put(pkId);
					log.error("从blockingQueue获取要发送的消息因为主线程事务阻塞,重新放回到blockingQueue中....pkId:{}",pkId);
				}
				else {
					String exchange = tbRabbitmqSendLog.getExchange();
					String routingKey = tbRabbitmqSendLog.getRoutingKey();
					
					MessageQueueBody messageQueueBody = new MessageQueueBody();
					messageQueueBody.setFlowNo(tbRabbitmqSendLog.getFlowNo());
					messageQueueBody.setJsonStr(tbRabbitmqSendLog.getJsonStr());
					
					String bodyStr = JSONObject.toJSONString(messageQueueBody);
					CorrelationData correlationData = new CorrelationData(tbRabbitmqSendLog.getFlowNo());
					Message message = MessageBuilder.withBody(bodyStr.getBytes()).build();
					
					rabbitTemplate.send(exchange,routingKey,message,correlationData);
					
					log.info("RabbitMQ发送消息队列完成Exchange:{},RoutingKey:{},pkId:{},flowNo:{}",
							tbRabbitmqSendLog.getExchange(),tbRabbitmqSendLog.getRoutingKey(),pkId,tbRabbitmqSendLog.getFlowNo());
					this.updateTbRabbitmqSendLog(pkId,true);
				}
			}
		}
		catch (Exception e) {
			log.error("RabbitMQ发送消息队列发生未知的异常pkId:"+pkId,e);
		}
	}
	
	
	@Override
	public boolean sendToExchange(String exchange,String routingKey,String jsonStr) {
		try {
			Long pkId = this.saveTbRabbitmqSendLog(exchange,routingKey,jsonStr);
			blockingQueue.put(pkId);
		}
		catch (Exception e) {
			log.error("发送消息出现未知的异常:"+e);
			return false;
		}
		
		return true;
	}
	
	@Override
	public boolean sendToQueue(String queueName,String jsonStr) {
		try {
			Long pkId = this.saveTbRabbitmqSendLog(null,queueName,jsonStr);
			blockingQueue.put(pkId);
		}
		catch (Exception e) {
			log.error("发送消息出现未知的异常:"+e);
			return false;
		}
		
		return true;
	}
	
	@Override
	public void resend() {
		Date now = new Date();
		Date beginTime = DateUtils.addMinutes(now,-120);
		Date endTime = DateUtils.addMinutes(now,-2);
		
		TbRabbitmqSendLog queryTbRabbitmqSendLog = new TbRabbitmqSendLog();
		queryTbRabbitmqSendLog.setSendStatus(false);
		queryTbRabbitmqSendLog.setExchangeStatus(false);
		queryTbRabbitmqSendLog.setRouteStatus(false);
		queryTbRabbitmqSendLog.setIsDeleted(false);
		List<TbRabbitmqSendLog> tbRabbitMqSendLogs = extTbRabbitmqSendLogMapper.selectExtListForResend(queryTbRabbitmqSendLog,beginTime,endTime);
		
		log.info("本次重发RabbitMQ失败的消息任务开始,总数目:{}",tbRabbitMqSendLogs.size());
		if(tbRabbitMqSendLogs.size() == 0) {
			return;
		}
		
		Integer successNum = 0;
		long taskBeginTime = System.currentTimeMillis();
		for(TbRabbitmqSendLog tbRabbitmqSendLog : tbRabbitMqSendLogs) {
			boolean success = true;
			try {
				this.resetQueueStatusForResend(tbRabbitmqSendLog);
				
				String exchange = tbRabbitmqSendLog.getExchange();
				String routingKey = tbRabbitmqSendLog.getRoutingKey();
				
				MessageQueueBody messageQueueBody = new MessageQueueBody();
				messageQueueBody.setFlowNo(tbRabbitmqSendLog.getFlowNo());
				messageQueueBody.setJsonStr(tbRabbitmqSendLog.getJsonStr());
				
				String bodyStr = JSONObject.toJSONString(messageQueueBody);
				CorrelationData correlationData = new CorrelationData(tbRabbitmqSendLog.getFlowNo());
				Message message = MessageBuilder.withBody(bodyStr.getBytes()).build();
				
				rabbitTemplate.send(exchange,routingKey,message,correlationData);
				
				log.info("RabbitMQ重新发送消息队列完成Exchange:{},RoutingKey:{},pkId:{},flowNo:{}",
						tbRabbitmqSendLog.getExchange(),tbRabbitmqSendLog.getRoutingKey(),tbRabbitmqSendLog.getId(),tbRabbitmqSendLog.getFlowNo());
				successNum++;
			}
			catch(Exception e) {
				success = false;
				log.error("本次重发RabbitMQ失败的消息任务发生未知异常..exception:"+e);
			}
			finally {
				this.updateSendStatusForResend(tbRabbitmqSendLog,success);
			}
		}
		
		log.info("本次重发MQ失败的消息任务结束,总数目:{},成功数目:{},耗时:{}",tbRabbitMqSendLogs.size(),successNum,(System.currentTimeMillis()-taskBeginTime)/1000);
	}
	
	@Override
	public boolean canQueueConsume(MessageQueueBody messageQueueBody,String consumerQueue) {
		//不能直接使用flowNo，一条消息可能会经exchange路由分发N个quque
		String newFlowNo = messageQueueBody.getFlowNo()+"#"+consumerQueue;
		
		TbRabbitmqConsumeLog insertTbRabbitmqConsumeLog = new TbRabbitmqConsumeLog();
		insertTbRabbitmqConsumeLog.setConsumerQueue(consumerQueue);
		insertTbRabbitmqConsumeLog.setFlowNo(newFlowNo);
		insertTbRabbitmqConsumeLog.setSuccess(true);
		insertTbRabbitmqConsumeLog.setJsonStr(messageQueueBody.getJsonStr());
		insertTbRabbitmqConsumeLog.setCreateTime(new Date());
		insertTbRabbitmqConsumeLog.setUpdateTime(new Date());
		insertTbRabbitmqConsumeLog.setIsDeleted(false);
		
		int resultCount = tbRabbitmqConsumeLogMapper.insertIgnoreSelective(insertTbRabbitmqConsumeLog);
		if(resultCount == 1) {
			//插入成功，可以消费
			return true;
		}
		
		return false;
	}
	
	/**
	 * 保存发送日志
	 * @param exchange
	 * @param routingKey
	 * @param jsonStr
	 * @return
	 */
	private Long saveTbRabbitmqSendLog(String exchange,String routingKey,String jsonStr) {
		String flowNo = UUID.randomUUID().toString().replaceAll("-", "");
		
		TbRabbitmqSendLog insertTbRabbitmqSendLog = new TbRabbitmqSendLog();
		insertTbRabbitmqSendLog.setExchange(exchange);
		insertTbRabbitmqSendLog.setRoutingKey(routingKey);
		insertTbRabbitmqSendLog.setFlowNo(flowNo);
		insertTbRabbitmqSendLog.setJsonStr(jsonStr);
		insertTbRabbitmqSendLog.setSendStatus(false);
		insertTbRabbitmqSendLog.setExchangeStatus(false);
		insertTbRabbitmqSendLog.setRouteStatus(true);
		insertTbRabbitmqSendLog.setRetryNum(0);
		insertTbRabbitmqSendLog.setCreateTime(new Date());
		insertTbRabbitmqSendLog.setUpdateTime(new Date());
		insertTbRabbitmqSendLog.setIsDeleted(false);			
		tbRabbitmqSendLogMapper.insertSelectiveReturnPrimaryKey(insertTbRabbitmqSendLog);
		
		return insertTbRabbitmqSendLog.getId();
	}
	/**
	 * 更新发送结果
	 * @param pkId
	 * @param success
	 */
	private void updateTbRabbitmqSendLog(Long pkId,boolean success) {
		TbRabbitmqSendLog updateTbRabbitmqSendLog = new TbRabbitmqSendLog();
		updateTbRabbitmqSendLog.setId(pkId);
		updateTbRabbitmqSendLog.setSendStatus(success);
		updateTbRabbitmqSendLog.setUpdateTime(new Date());
		
		tbRabbitmqSendLogMapper.updateByPrimaryKeySelective(updateTbRabbitmqSendLog);
	}
	/**
	 * queue_status只有在失败的时候，才会重发回调；重试前重置为true
	 * @param tbRabbitmqSendLog
	 */
	private void resetQueueStatusForResend(TbRabbitmqSendLog tbRabbitmqSendLog) {
		if(!tbRabbitmqSendLog.getRouteStatus()) {
			TbRabbitmqSendLog updateTbRabbitmqSendLog = new TbRabbitmqSendLog();
			updateTbRabbitmqSendLog.setId(tbRabbitmqSendLog.getId());
			updateTbRabbitmqSendLog.setRouteStatus(true);
			updateTbRabbitmqSendLog.setUpdateTime(new Date());
			
			tbRabbitmqSendLogMapper.updateByPrimaryKeySelective(updateTbRabbitmqSendLog);
		}
	}
	/**
	 * 重新发送后更新send_status，retry_num + 1
	 * @param tbRabbitmqSendLog
	 * @param success
	 */
	private void updateSendStatusForResend(TbRabbitmqSendLog tbRabbitmqSendLog,boolean success) {
		TbRabbitmqSendLog updateTbRabbitmqSendLog = new TbRabbitmqSendLog();
		updateTbRabbitmqSendLog.setId(tbRabbitmqSendLog.getId());
		updateTbRabbitmqSendLog.setSendStatus(success);
		updateTbRabbitmqSendLog.setRetryNum(tbRabbitmqSendLog.getRetryNum() + 1);
		updateTbRabbitmqSendLog.setUpdateTime(new Date());
		
		tbRabbitmqSendLogMapper.updateByPrimaryKeySelective(updateTbRabbitmqSendLog);
	}
}